package xyz.xiaomeiqiu.netty.outbound;

import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOutboundHandlerAdapter;
import io.netty.handler.codec.http.*;
import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.concurrent.FutureCallback;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.apache.http.protocol.HTTP;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import xyz.xiaomeiqiu.netty.filter.HeaderHttpResponseFilter;
import xyz.xiaomeiqiu.netty.filter.HttpRequestFilter;
import xyz.xiaomeiqiu.netty.filter.HttpResponseFilter;
import xyz.xiaomeiqiu.netty.router.HttpEndpointRouter;
import xyz.xiaomeiqiu.netty.router.RandomHttpEndpointRouter;

import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.stream.Collectors;

import static io.netty.handler.codec.http.HttpResponseStatus.OK;
import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;

public class HttpOutboundHandler {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    private final List<String> proxyServer;
    private final CloseableHttpAsyncClient httpAsyncClient;
    private final ExecutorService proxyService;

    private final HttpResponseFilter filter = new HeaderHttpResponseFilter();

    private final HttpEndpointRouter router = new RandomHttpEndpointRouter();

    public HttpOutboundHandler(List<String> proxyServer){
        logger.info("HttpOutboundHandler 初始化");
        this.proxyServer = proxyServer.stream().map(this::formatUrl).collect(Collectors.toList());

        int cores = Runtime.getRuntime().availableProcessors();
        long keepAliveTime = 1000;
        int queueSize = 2048;

        RejectedExecutionHandler handler = new ThreadPoolExecutor.CallerRunsPolicy();//.DiscardPolicy();
        proxyService = new ThreadPoolExecutor(cores, cores,
                keepAliveTime, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(queueSize),
                new NamedThreadFactory("proxyService"), handler);

        IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
                .setConnectTimeout(1000)
                .setSoTimeout(1000)
                .setIoThreadCount(cores)
                .setRcvBufSize(32 * 1024).build();
        httpAsyncClient = HttpAsyncClients.custom()
                .setMaxConnTotal(40)
                .setMaxConnPerRoute(8)
                .setDefaultIOReactorConfig(ioReactorConfig)
                .setKeepAliveStrategy(((httpResponse, httpContext) -> 6000)).build();
        httpAsyncClient.start();
        logger.info(" httpAsyncClient 已启动 ");
    }

    private String formatUrl(String url){
        return url.endsWith("/")?url.substring(0,url.length()-1):url;
    }

    public void handle(FullHttpRequest inbound, ChannelHandlerContext ctx, HttpRequestFilter filter) {
        String backendUrl = router.route(this.proxyServer);
        final String url = backendUrl + inbound.uri();
        logger.info(" 选定后端地址:{} ", url);
        filter.filter(inbound, ctx);
        proxyService.submit(()->fetchGet(inbound, ctx, url));
    }

    private void fetchGet(FullHttpRequest inbound, ChannelHandlerContext ctx, String url) {
        final HttpGet httpGet = new HttpGet(url);
        httpGet.setHeader(HTTP.CONN_DIRECTIVE, HTTP.CONN_KEEP_ALIVE);

        for (Map.Entry<String, String> header : inbound.headers()) {
            httpGet.setHeader(header.getKey(), header.getValue());
        }

        httpAsyncClient.execute(httpGet, new FutureCallback<HttpResponse>() {
            @Override
            public void completed(HttpResponse httpResponse) {
                try {
                    handleResponse(inbound, ctx, httpResponse);
                } catch (Exception e) {
                    logger.error("处理接口返回结果失败", e);
                    e.printStackTrace();
                }
            }

            @Override
            public void failed(Exception e) {
                httpGet.abort();
                logger.error("执行http请求失败 ", e);
            }

            @Override
            public void cancelled() {
                logger.warn("请求 cancelled");
                httpGet.abort();
            }
        });
    }

    private void handleResponse(final FullHttpRequest fullRequest, final ChannelHandlerContext ctx, final HttpResponse endpointResponse) {
        FullHttpResponse response = null;
        try {
            byte[] body = EntityUtils.toByteArray(endpointResponse.getEntity());
            response = new DefaultFullHttpResponse(HTTP_1_1, OK, Unpooled.wrappedBuffer(body));
            response.headers().set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON);
            response.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, Integer.parseInt(endpointResponse.getFirstHeader(HttpHeaderNames.CONTENT_LENGTH.toString()).getValue()));

            filter.filter(response);

        } catch (Exception e){
            exceptionCaught(ctx, e);
        } finally {
            if (fullRequest != null) {
                if (!HttpUtil.isKeepAlive(fullRequest)) {
                    ctx.write(response).addListener(ChannelFutureListener.CLOSE);
                } else {
                    //response.headers().set(CONNECTION, KEEP_ALIVE);
                    ctx.write(response);
                }
            }
            ctx.flush();
        }
    }

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        logger.error("发生错误", cause);
        ctx.close();
    }
}
